package com.health.trans.channel;

import org.apache.commons.lang.Validate;

import java.util.ArrayList;
import java.util.List;

/**
 * @author nqlz
 */
public class BufferedRecordExchanger<T> {

    private final Channel<T> channel;

    private final List<T> buffer;

    private int bufferSize;

    private int bufferIndex = 0;

    private volatile boolean shutdown = false;


    @SuppressWarnings("unchecked")
    public BufferedRecordExchanger(final Channel<T> channel) {
        assert null != channel;
        this.channel = channel;
        this.bufferSize = 2048;
        this.buffer = new ArrayList<>(bufferSize);
    }


    public void sendToWriter(T record) {
        Validate.notNull(record, "record不能为空.");
        boolean isFull = (this.bufferIndex >= this.bufferSize);
        if (isFull) {
            flush();
        }
        this.buffer.add(record);
        this.bufferIndex++;
    }

    public void flush() {
        this.channel.pushAll(this.buffer);
        this.buffer.clear();
        this.bufferIndex = 0;
    }


    public T getFromReader() {
        boolean isEmpty = (this.bufferIndex >= this.buffer.size());
        if (isEmpty) {
            receive();
        }
        return this.buffer.get(this.bufferIndex++);
    }

    public void shutdown() {
        shutdown = true;
        try {
            buffer.clear();
            channel.clear();
        } catch (Throwable t) {
            t.printStackTrace();
        }
    }

    private void receive() {
        this.channel.pullAll(this.buffer);
        this.bufferIndex = 0;
        this.bufferSize = this.buffer.size();
    }

}
